package rpc.client.netty;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.pojo.Instance;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import rpc.common.enumeraion.CompressType;
import rpc.common.enumeraion.PackageType;
import rpc.common.enumeraion.SerializerCode;
import rpc.common.utils.NacosUtil;
import rpc.common.utils.SingletonFactory;
import rpc.intf.RpcClient;
import rpc.intf.LoadBalancer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import rpc.packet.RpcMessage;
import rpc.packet.RpcRequest;
import rpc.packet.RpcResponse;
import rpc.packet.codec.CommonCodec;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;


@Slf4j
@Data
public class NettyClient implements RpcClient {

    private static final Bootstrap bootstrap;
    private static final EventLoopGroup group;
    private static final UnprocessedRequests unprocessedRequests;

    //负载均衡算法
    private LoadBalancer loadBalancer;

    public NettyClient(LoadBalancer loadBalancer) {
        this.loadBalancer = loadBalancer;
    }

    static  {
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                //设置客户端通道实现类型
                .channel(NioSocketChannel.class)
                //option给SocketChannel配置参数
                //.option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)//5秒 连接超时
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(
                                1024, 7, 4, 0, 0));
                        pipeline.addLast(new LoggingHandler(LogLevel.DEBUG))
                                .addLast(new CommonCodec())
                                .addLast(new NettyClientHandler())
                        ;
                    }
                });
        unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
    }


    @Override
    public Object sendRequest(RpcRequest rpcRequest) {
        //返回结果
        CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>();
        log.debug("服务: {}", rpcRequest.getInterfaceName());
        //获取提供该服务的 服务器ip
        InetSocketAddress ipAndPort = lookupService(rpcRequest.getInterfaceName());

        RpcMessage rpcMessage = RpcMessage.builder()
                .packageType(PackageType.REQUEST_PACK.getCode())
                .serializer(SerializerCode.KRYO.getCode())
                .compress(CompressType.GZIP.getCode())
                .requestId(rpcRequest.getRequestId())
                .build();
        try {
            ChannelFuture future = bootstrap.connect(ipAndPort).sync();//同步等待
            log.debug("客户端连接到服务器 {}", ipAndPort.toString());
            Channel channel = future.channel();
            if(channel != null) {
                rpcMessage.setData(rpcRequest);
                channel.writeAndFlush(rpcMessage).addListener(promise -> {
                    if(promise.isSuccess()) {
                        log.debug(String.format("客户端发送消息: %s", rpcMessage));
                        //加入集合，等待结果返回
                        unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture);
                    } else {
                        channel.close();
                        resultFuture.completeExceptionally(promise.cause());
                        log.error("发送消息时有错误发生: ", promise.cause());
                    }
                });
            }
            //调用一次服务就关闭，短连接。
            //对通道关闭进行监听
//            channel.closeFuture().addListener(future1 -> {
//                log.debug("优雅关闭");
//                group.shutdownGracefully();
//            });
        } catch (InterruptedException e) {
            log.error("发送消息时有错误发生: ", e);
        }
        return resultFuture;
    }

    /**
     * 根据相应的负载均衡算法，选取服务器
     * @param serviceName
     * @return
     */
    private InetSocketAddress lookupService(String serviceName) {
        try {
            //获取提供该服务的机器 ip 集合
            List<Instance> instances = NacosUtil.getAllInstances(serviceName);
            //负载均衡策略
            //一致性哈希算法
            Instance serverInfo = loadBalancer.select(instances);
            log.debug("选取服务器...");
            return new InetSocketAddress(serverInfo.getIp(), serverInfo.getPort());
        } catch (NacosException e) {
            log.debug("获取服务器时有错误发生");
        }
        return null;
    }
}
